#include <stdlib.h>
#include <stdio.h>

#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
#include <time.h>
#include <sys/neutrino.h>
#include <inttypes.h>
#include <string.h>

#include <cogent.h>

#include "filedb.h"
#include "memdb.h"

//#define DEBUG_PLCDB

histories_t		histories;

//----------------------------------------------------------------------------
// flush in-memory buffers to disk and empty them
//----------------------------------------------------------------------------
int flush (cfg_t *cfg)
{
	int fd;
	char namebuf [512];
	
	if (cfg->filedb && cfg->filedb->last) {
		if ((cfg->old->npt > 0) || (cfg->new->npt > 0)) {
			sprintf (namebuf, "%s/%s", cfg->filedb->base_dir, cfg->filedb->last->name);
			#ifdef DEBUG_PLCDB
			printf ("flush buffers to file \"%s\"\n", namebuf);
			#endif
			fd = open (namebuf, O_WRONLY | O_CREAT, S_IFREG);
			lseek (fd, 0, SEEK_END);
			write (fd, cfg->old->pt, cfg->old->npt * sizeof (pt_t)); 
			write (fd, cfg->new->pt, cfg->new->npt * sizeof (pt_t)); 
			if (cfg->new->npt > 0) {
				if (cfg->filedb->last->beg_pt.t > 0) {
					memcpy (&cfg->filedb->last->end_pt, &cfg->new->pt [cfg->new->npt -1], sizeof (pt_t));
				} else {
					if (cfg->old->npt > 0) {
						memcpy (&cfg->filedb->last->beg_pt, &cfg->old->pt [0], sizeof (pt_t));
						memcpy (&cfg->filedb->last->end_pt, &cfg->new->pt [cfg->new->npt -1], sizeof (pt_t));
					} else {
						memcpy (&cfg->filedb->last->beg_pt, &cfg->new->pt [0], sizeof (pt_t));
						memcpy (&cfg->filedb->last->end_pt, &cfg->new->pt [cfg->new->npt -1], sizeof (pt_t));				
					}
				}
			} else {
				if (cfg->old->npt > 0) {
					if (cfg->filedb->last->beg_pt.t > 0) {
						memcpy (&cfg->filedb->last->end_pt, &cfg->old->pt [cfg->old->npt -1], sizeof (pt_t));
					} else {			
						memcpy (&cfg->filedb->last->beg_pt, &cfg->old->pt [0], sizeof (pt_t));
						memcpy (&cfg->filedb->last->end_pt, &cfg->old->pt [cfg->old->npt -1], sizeof (pt_t));				
					}
				}
			}
		
			close (fd);
			cfg->new->npt = 0;
			cfg->old->npt = 0;
		}
	}
	
	return 0;	
}

//----------------------------------------------------------------------------
// start new file
//----------------------------------------------------------------------------
int start_new_file (cfg_t *cfg) 
{
	char namebuf [256];
	file_t *file;
	filedb_t *filedb;
	
	filedb = cfg->filedb;	
	if (filedb) {
		new_fname (namebuf, filedb->base_name, "dat", time (NULL));
		if (filedb->last) {
			if (!strcmp (filedb->last->name, namebuf)) {
				// уже есть файл с таким именем
				return -1;	
			} else {
				file = malloc (sizeof (file_t));
				memset (file, 0, sizeof (file_t));
				filedb->last->next = file;
				file->prev = filedb->last;
				filedb->last = file;
				file->name = strdup (namebuf);					
			}
		} else {
			file = malloc (sizeof (file_t));
			memset (file, 0, sizeof (file_t));
			filedb->last = file;
			filedb->first = file;
			file->name = strdup (namebuf);			
		}
		#ifdef DEBUG_PLCDB
		printf ("new filename = \"%s\"\n", namebuf);
		#endif	
		flush (cfg);
	}			

	return 0;	
}

//----------------------------------------------------------------------------
// add element to current buffer
//----------------------------------------------------------------------------
int add (cfg_t *cfg, pt_t *pt)
{
	buf_t *buf;
	int fd;
	char namebuf [512];
	
	#ifdef DEBUG_PLCDB
	printf ("adding point \"%s\" [%d] = %d (t = %f)\n", cfg->point_name, cfg->new->npt, pt->y, pt->t);
	printf ("buf addr = %p, size = %d\n", cfg->new->pt, cfg->new->npt);
	#endif
	cfg->new->pt [cfg->new->npt++] = *pt;	
	if (cfg->new->npt >= cfg->buf_size) {
		buf = cfg->new;
		if (cfg->old->npt > 0) {
			if (cfg->filedb && (cfg->filedb->last == NULL)) start_new_file (cfg);
			if (cfg->filedb && cfg->filedb->last) {
				if (cfg->old->npt > 0) {
					sprintf (namebuf, "%s/%s", cfg->filedb->base_dir, cfg->filedb->last->name);
					#ifdef DEBUG_PLCDB
					printf ("writing to file \"%s\"\n", namebuf);
					#endif
					fd = open (namebuf, O_WRONLY | O_CREAT, S_IFREG);
					lseek (fd, 0, SEEK_END);
					write (fd, cfg->old->pt, cfg->old->npt * sizeof (pt_t)); 
					if (cfg->filedb->last->beg_pt.t > 0) {
						memcpy (&cfg->filedb->last->end_pt, &cfg->old->pt [cfg->old->npt -1], sizeof (pt_t));
					} else {
						memcpy (&cfg->filedb->last->beg_pt, &cfg->old->pt [0], sizeof (pt_t));
						memcpy (&cfg->filedb->last->end_pt, &cfg->old->pt [cfg->old->npt -1], sizeof (pt_t));					
					}
					close (fd);
				}
			}
		}		
		cfg->new = cfg->old;
		cfg->new->npt = 0;
		cfg->old = buf;
	}
	
	return 0; 
}

//----------------------------------------------------------------------------
// add history to histories array
//----------------------------------------------------------------------------
int add_history (histories_t *histories, char *point_name)
{
	cfg_t *cfg;
	char namebuf [512];	

	histories->ncfg ++;
	histories->cfg = realloc (histories->cfg, histories->ncfg * sizeof (cfg_t));	
	if (histories->cfg == NULL) {
		fprintf (stderr, "can't realloc in add_history");
		exit (EXIT_FAILURE);
	}
	cfg = &histories->cfg [histories->ncfg -1];
	memset (cfg, 0, sizeof (cfg_t));
	//printf ("size = %d\n", sizeof (cfg_t));
	//printf ("size = %d\n", sizeof (histories->cfg));
	
	cfg->buf_size = histories->buf_size;
	init_memdb (cfg);
	//printf ("init buf->new->npt = %d\n", cfg->new->npt);

	cfg->point_name = strdup (point_name);
	cfg->filedb = malloc (sizeof (filedb_t));
	memset (cfg->filedb, 0, sizeof (filedb_t));
	
	sprintf (namebuf, "%s/%s", histories->base_dir, point_name);
	cfg->filedb->base_dir  = strdup (namebuf);
	cfg->filedb->base_name = strdup (point_name);
	mkdir (namebuf, S_IFREG);
	
	//enum_files (cfg->filedb);
	//#ifdef DEBUG_TRENDDB
	//printf ("======= enum_files ========\n");
	//print_enum_files (cfg->filedb);
	//#endif
	
	//start_new_file (cfg);

	return 0;	
}

//----------------------------------------------------------------------------
// enum all history files
//----------------------------------------------------------------------------
int enum_history (cfg_t *cfg)
{
	filedb_t *filedb;
	file_t *cur, *next;
	
	filedb = cfg->filedb;
	if (filedb) {		
		if (filedb->first) {
			cur = filedb->first;
			while (cur) {
				next = cur->next;
				free (cur->name);
				free (cur);
				cur = next;
			}
			filedb->first = NULL;
			filedb->last = NULL;
		}
		enum_files (cfg->filedb);
	}
	#ifdef DEBUG_PLCDB
	printf ("======= enum_files ========\n");
	print_enum_files (cfg->filedb);
	#endif	
	
	return 0;	
}

//----------------------------------------------------------------------------
// query points from archive
//----------------------------------------------------------------------------
int query (cfg_t *cfg, double t_beg, double t_end, pt_t **values, int *nvalues)
{
	int end_found_1 = 0;
	int end_found_2 = 0;
	pt_t *buf_1 = NULL;
	pt_t *buf_2 = NULL;
	pt_t *out   = NULL;
	int buf_1_size = 0;
	int buf_2_size = 0;
	int out_size   = 0;
	pt_t pt;
	
	end_found_1 = 0;
	if ((memdb_first_pt (cfg, &pt) == 0) && (pt.t <= t_beg)) {
		buf_1_size = 0;
		buf_1 = NULL;
		end_found_2 = query_memdb (cfg, t_beg, t_end, &buf_2, &buf_2_size);		

	} else {
		if (cfg->filedb) {
			end_found_1 = query_files (cfg->filedb, t_beg, t_end, &buf_1, &buf_1_size);
			//printf ("end_found_1 = %d, buf_1_size = %d\n", end_found_1, buf_1_size);
		}
		if (end_found_1 == 0) {
			end_found_2 = query_memdb (cfg, t_beg, t_end, &buf_2, &buf_2_size);
			//printf ("end_found_2 = %d, buf_2_size = %d\n", end_found_2, buf_2_size);
		}
	}
	
	out = NULL;
	out_size = 0;
	if (buf_1_size > 0) {
		out = buf_1;	
		out_size = buf_1_size;
		if (buf_2_size > 0) {
			out_size += buf_2_size;
			out = realloc (out, out_size * sizeof (pt_t));
			memcpy (&out [buf_1_size], buf_2, buf_2_size * sizeof (pt_t));
			free (buf_2);
			buf_2 = NULL;
			buf_2_size = 0;
		}

	} else if (buf_2_size > 0) {
		out = buf_2;
		out_size = buf_2_size;
	}
	
	*values  = out;
	*nvalues = out_size;
	//printf ("out_size = %d\n", out_size);
	
	return 0;
}

//----------------------------------------------------------------------------
// process point and save it for history
//----------------------------------------------------------------------------
int process_history (cfg_t *cfg, PT_pCPOINT ppoint)
{
	pt_t pt;
	
	//#ifdef DEBUG_PLCDB
	//printf ("entering process_history \"%s\"..\n", cfg->point_name);
	//#endif
	if (strcmp (cfg->point_name, ppoint->name)) {
		return -1;
	}
	#ifdef DEBUG_PLCDB
	printf ("processing history \"%s\"..\n", cfg->point_name);
	#endif
	
	switch (ppoint->type) {
		case PT_TYPE_INT32:
			pt.t  = ppoint->seconds;
			pt.t += ppoint->nanoseconds / 1000000000.0;
			pt.y  = ppoint->value.i;
			if (pt.t > 0) {
				add (cfg, &pt);
			}
			break;
		case PT_TYPE_REAL:
			pt.t  = ppoint->seconds;
			pt.t += ppoint->nanoseconds / 1000000000.0;
			pt.y  = ppoint->value.r;
			if (pt.t > 0) {
				add (cfg, &pt);
			}
			break;
		case PT_TYPE_STRING:
			return -1;
			break;
		default:
			return -1;
			break;
	}	
	
	return 0;
}

//----------------------------------------------------------------------------
// register programm paths and histories
//----------------------------------------------------------------------------
int register_historian (histories_t *histories, IP_Msg **phmsg, IP_Task **phtask)
{
	int				i;
	IP_Msg			*hmsg;
	ST_STATUS		status;
	PT_stCPOINT		point;
	char			namebuf [256];
	char			*qname = NULL, *domain = NULL;
	char			*name = NULL, *myname;
	IP_Task			*htask = NULL;
	
	hmsg = IP_MsgCreate (NULL, IP_MsgDefaultSize(), 0);
	sprintf (namebuf, "sc/wait%d", getpid());
	qname = strdup (namebuf);
	sprintf (namebuf, "sc/wait%d", getpid());
	name = strdup (namebuf);
	myname = strdup ("plcdb");
	
	if (!(htask = IP_NserveInit (myname, domain, qname, 0, 0))) {
		printf ("IP_NserveInit () failed: are qserve and nserve running?\n");
		exit (EXIT_FAILURE);
	}

	// register all histories
	for (i= 0; i< histories->ncfg; i++) {
		memset (&point, 0, sizeof (point));
		point.name = histories->cfg [i].point_name;
		if ((status = DH_RegisterPoint (htask, &point, hmsg, NULL)) != ST_OK) {
			printf ("Register point failed: %d\n", status);	
			return -1;
		}
		process_history (&histories->cfg [i], &point);
		#ifdef DEBUG_PLCDB
		printf ("registered point \"%s\"\n", point.name);
		#endif		
	}

	*phmsg  = hmsg;
	*phtask = htask;
	return 0;	
}

//----------------------------------------------------------------------------
// buffer_count_gaps
//----------------------------------------------------------------------------
int buffer_count_gaps (query_buf_t *query_buf, double t_end, double gap, int *ngaps)
{	
	int i;
	pt_t *pt_prev, *pt_next;

	*ngaps = 0;
	if (query_buf) {
		if (query_buf->nvalues > 0) {
			pt_prev = &query_buf->values [0];
			for (i= 1; i< query_buf->nvalues; i++) {
				//pt_prev = &query_buf->values [i -1];
				pt_next = &query_buf->values [i];
				if ((pt_next->t - pt_prev->t) > gap) {
					(*ngaps) ++;
				}			
				pt_prev = pt_next;
			}
			if (query_buf->values [query_buf->nvalues -1].t < t_end) {
				(*ngaps) ++;	
			}
		}
	}
	
	#ifdef DEBUG_PLCDB
	printf ("buffer_count_gaps - gaps = %d\n", *ngaps);
	#endif
		
	return 0;	
}

//----------------------------------------------------------------------------
// buffer_fill_gaps
//----------------------------------------------------------------------------
int buffer_fill_gaps (query_buf_t *query_buf, double t_end, double gap, pt_t *new_pt, int *nnew_pt)
{	
	int i;
	pt_t *pt_prev, *pt_next;
	int npt;

	npt = 0;
	if (query_buf) {
		if (query_buf->nvalues > 0) {
			pt_prev = &query_buf->values [0];
			new_pt [npt ++] = *pt_prev;			
			for (i= 1; i< query_buf->nvalues; i++) {
				pt_next = &query_buf->values [i];
				if ((pt_next->t - pt_prev->t) > gap) {
					new_pt [npt].y = pt_prev->y;
					new_pt [npt].t = pt_next->t;
					npt ++;
				}
				new_pt [npt ++] = *pt_next;			
				//if (i == (query_buf->nvalues -1)) {
				//	new_pt [npt ++] = *pt_next;
				//}
				pt_prev = pt_next;
			}
			if (query_buf->values [query_buf->nvalues -1].t < t_end) {
				new_pt [npt].y = query_buf->values [query_buf->nvalues -1].y;
				new_pt [npt].t = t_end;
				npt ++;			
			}
		}
	}
		
	*nnew_pt = npt;
	return 0;	
}

//----------------------------------------------------------------------------
// interpolate
//----------------------------------------------------------------------------
int interpolate (int *buffer_id, double t_beg, double t_end, char *name, histories_t *histories)
{
	int i;
	pt_t *values;
	int nvalues;
	cfg_t *cfg;
	int ret;
	double gap;
	int ngaps;
	pt_t *new_pt;
	int nnew_pt;
	
	query_buf_t *query_buf = NULL;
	
	for (i= 0; i< histories->ncfg; i++) {
		cfg = &histories->cfg [i];
		if (!strcmp (cfg->point_name, name)) {
			break;
		}
	}
	if (i >= histories->ncfg) {
		printf ("can't find history with name \"%s\"\n", name);	
		return -1;
	}
	
	ret = query (cfg, t_beg, t_end, &values, &nvalues);
	if (ret == 0) {
		query_buf = malloc (sizeof (query_buf_t));
		memset (query_buf, 0, sizeof (query_buf_t));
		//printf ("*values = %p\n", query_buf->values);
		//if (histories->query_buf.values) {
		//	printf ("hello");
		//	free (histories->query_buf.values);
		//}
		query_buf->values = values;
		query_buf->nvalues = nvalues;
		query_buf->t_beg = t_beg;
		query_buf->t_end = t_end;
	}

	gap = 2.0;
	if (gap > 0) {
		buffer_count_gaps (query_buf, t_end, gap, &ngaps);
		//printf ("ngaps = %d\n", ngaps);
		if (ngaps > 0) {
			new_pt = malloc ((query_buf->nvalues + ngaps) * sizeof (pt_t));
			buffer_fill_gaps (query_buf, t_end, gap, new_pt, &nnew_pt);
			#ifdef DEBUG_PLCDB
			printf ("nold_pt = %d, nnew_pt = %d\n", query_buf->nvalues, nnew_pt);
			#endif
			if (query_buf->values) free (query_buf->values);
			query_buf->values = new_pt;
			query_buf->nvalues = nnew_pt;
			//if (nnew_pt > 0) {
				//printf ("nnew_pt = %d, new_pt [0].y  = %f\n", nnew_pt, new_pt [0].y);	
			//}
		}
	}
	
	*buffer_id = (int) query_buf;
	return 0;	
}

//----------------------------------------------------------------------------
// buffer_id_lenght
//----------------------------------------------------------------------------
int buffer_id_lenght (int buffer_id, int *lenght, histories_t *histories)
{	
	if (buffer_id) {
		query_buf_t *query_buf = (query_buf_t *) buffer_id;		
		*lenght = query_buf->nvalues;
	}
		
	return 0;	
}

//----------------------------------------------------------------------------
// buffer_id_read
//----------------------------------------------------------------------------
int buffer_id_read (int buffer_id, int start, int count, histories_t *histories, char *buf, int *buflen)
{
	if (buffer_id) {
		query_buf_t *query_buf = (query_buf_t *) buffer_id;		
		if ((start + count) > query_buf->nvalues) {
			return -1;	
		}
		if (count > ((*buflen) / sizeof (pt_t))) {
			printf ("too small buffer for buffer_id_read\n");
			return -1;
		}
	
		*buflen  = min (count, (query_buf->nvalues - start));
		*buflen *= sizeof (pt_t);
		memcpy (buf, &query_buf->values [start], *buflen);
	}
		
	return 0;	
}

//----------------------------------------------------------------------------
// buffer_id_destroy
//----------------------------------------------------------------------------
int buffer_id_destroy (int buffer_id, histories_t *histories)
{
	if (buffer_id) {
		query_buf_t *query_buf = (query_buf_t *) buffer_id;		
		#ifdef DEBUG_PLCDB
		printf ("*values = %p\n", query_buf->values);
		#endif
	
		if (query_buf->values) {
			free (query_buf->values);
			query_buf->values = NULL;
		}
		query_buf->nvalues = 0;
		query_buf->t_beg = 0;
		query_buf->t_end = 0;
	
		#ifdef DEBUG_PLCDB
		printf ("hello from buffer_id_destroy end\n");
		#endif
		free (query_buf);
	}

	return 0;	
}

//----------------------------------------------------------------------------
// signal handler for SIGTERM
//----------------------------------------------------------------------------
void signal_handler (int sig)
{
	cfg_t *cfg;
	pt_t pt;
	struct timespec	tp;
	int i;	

	clock_gettime (CLOCK_REALTIME, &tp);	
	pt.y = 0;
	pt.t = tp.tv_sec + tp.tv_nsec /1000000000.0;
	for (i =0; i< histories.ncfg; i++) {
		cfg = &histories.cfg [i];
		add (cfg, &pt);
		flush (cfg);
		printf ("cfg->point_name = '%s'\n", cfg->point_name);
	}	
	
   	_exit (1);
}

//----------------------------------------------------------------------------
// main
//----------------------------------------------------------------------------
int main (int argc, char *argv[]) 
{
	IP_Msg			*hmsg;
	IP_Task			*htask;
	IP_MsgInfo		msginfo;
	int				type;
	//histories_t		histories;
	PT_stCPOINT		point;
	char			*msgend;
	char			namebuf [256];
	int				i, j;
	int				ret;
	char			buf [2000];
	int				buflen;
	char			base_dir [] = "/histories/plc"; 
	
	char			*pt_name_base = "CJ1M";
	char			*pt_name_type [] = 	{	"CIO_WRD",	"CIO_WRD",
											"WR_WRD",	"TIM_WRD",
											"TIM_BIT",	"CNT_WRD",
											"CNT_BIT" };
	int				pt_name_start [] = 	{ 0, 2000, 0, 0, 0, 0, 0};
	int				pt_name_size [] = 	{ 2, 20, 55, 40, 40, 20, 20};
	
	//	<mem name="CIO_WRD"		addr="0" 	n_words="2">	</mem>
	//	<mem name="CIO_WRD"		addr="2000"	n_words="20">	</mem>
	//	<mem name="WR_WRD"		addr="0" 	n_words="55">	</mem>
	//	<mem name="TIM_WRD"		addr="0" 	n_words="40">	</mem>
	//	<mem name="TIM_BIT"		addr="0"	n_words="40">	</mem>
	//	<mem name="CNT_WRD"		addr="0"	n_words="20">	</mem>
	//	<mem name="CNT_BIT"		addr="0"	n_words="20">	</mem>
	
	int				n_pt_name_type = sizeof (pt_name_type)/ sizeof (*pt_name_type);

	memset (&histories, 0, sizeof (histories));

	mkdir (base_dir, S_IFREG);
	histories.base_dir = strdup (base_dir);
	histories.buf_size = 200;
	for (i= 0; i< n_pt_name_type; i++) {
		for (j= 0; j< pt_name_size [i]; j++) {
			sprintf (namebuf, "%s_%s_%06d", pt_name_base, pt_name_type [i], pt_name_start [i] + j);
			add_history (&histories, namebuf);			
		}
	}
	register_historian (&histories, &hmsg, &htask);
	
	for (i= 0; i< histories.ncfg; i++) {
		enum_history (&histories.cfg [i]);
		start_new_file (&histories.cfg [i]);
	}
	
	memset (&point, 0, sizeof (point));
	point.name = namebuf;

	//signal (SIGTERM, signal_handler);
	
	// Infinite loop for message processing
	for (;;) {
		type = IP_Receive (htask, hmsg, &msginfo);
		#ifdef DEBUG_PLCDB
		printf ("Received: %s\n", (char*)IP_MsgData (hmsg));
		#endif

		switch (type) {
			case IP_ASYNC:
				switch (IP_MsgSubtype (hmsg)) {
					case ST_DH_EXCEPTION:
						for (msgend = IP_MsgData (hmsg); *msgend; ) {
							DH_ParsePointString (&point, point.name, msgend, &msgend, NULL);
							for (i= 0; i< histories.ncfg; i++) {
								process_history (&histories.cfg [i], &point);						
							}
						}
						break;
					case ST_DH_ECHO:
						/* We never write a point, so ECHO is impossible in this
						 * app.  If we expected echoes, we could handle them in
						 * the same way as exceptions.  The message contents are
						 * identical for both. */
						break;
			 		 default:
						/* Async general ASCII message, sent by another task using
						 * the Cascade DataHub API.  One such task is the Cascade
						 * name server, nserve.  It will send taskstarted and
						 * taskdied messages when other tasks start and stop:
						 * (taskstarted <name> <domain> <queue> <node> <pid> <chid>)
						 * (taskdied <name> <domain> <queue> <node> <pid> <chid>)
						 */
						break;
				}
				break;
			case IP_ERROR:
				/* Receive returned zero, which should never happen. */
				break;
			case IP_SYSTEM:
				/* A task death message that can only be received by a task
				 * that has set its QNX INFORMED bit.  You should avoid this,
				 * and use the taskdied and taskstarted messages in the
				 * IP_ASYNC section if possible. */
				break;
		  	case IP_SIGNAL:
				/* A signal caused the Receive to exit prematurely.  No message
			 	 * was received.  The value of sender is not defined. */
				break;
		  	case IP_NONE:
				/* The queue reported a message waiting, but none was available.
				 * This happens if the application intentionally drains the queue,
				 * which is generally a bad idea. */
				break;
			case IP_SYNC:
				/* Synchronous general ASCII message, sent by another task using
				 * the Cascade DataHub API. */
				// создать буфер данных (interpolate [t_beg] [t_end] [history_name])
				if (!strncmp ((char*) hmsg->msg->data, "interpolate", 11)) {
					double t_beg, t_end;
					int buffer_id;					
					sscanf ((char*) hmsg->msg->data + 11, "%lf %lf %s", &t_beg, &t_end, buf);
					#ifdef DEBUG_PLCDB
					printf ("interpolate (%f to %f) for \"%s\"\n", t_beg, t_end, buf);					
					#endif
					ret = interpolate (&buffer_id, t_beg, t_end, buf, &histories);
					if (ret == 0) {
						buflen = sprintf (buf, "%d", buffer_id) + 1;
					} else {
						buflen = sprintf (buf, "ERROR") + 1;
					}						
					IP_MsgCascade (hmsg, buf, buflen, IP_NONE, ST_OK);
					IP_MsgInfoReply (&msginfo, hmsg);
					
				} else
				// определить длину буфера данных (buffer_id_lenght [id])
				if (!strncmp ((char*) hmsg->msg->data, "buffer_id_lenght", 16)) {
					int buffer_id;
					int len;
					sscanf ((char*) hmsg->msg->data + 16, "%d", &buffer_id);
					#ifdef DEBUG_PLCDB
					printf ("buffer_id_lenght (id = %d)\n", buffer_id);
					#endif
					ret = buffer_id_lenght (buffer_id, &len, &histories);
					if (ret == 0) {
						buflen = sprintf (buf, "%d", len) + 1;
					} else {
						buflen = sprintf (buf, "ERROR") + 1;
					}						
					IP_MsgCascade (hmsg, buf, buflen, IP_NONE, ST_OK);
					IP_MsgInfoReply (&msginfo, hmsg);
					
				} else
				// прочитать буфер данных (buffer_id_read [id] [start] [count])
				if (!strncmp ((char*) hmsg->msg->data, "buffer_id_read", 14)) {
					int buffer_id;
					int start;
					int count;
					sscanf ((char*) hmsg->msg->data + 14, "%d %d %d", &buffer_id, &start, &count);
					#ifdef DEBUG_PLCDB
					printf ("buffer_id_read (id = %d) start = %d, count = %d\n", buffer_id, start, count);
					#endif
					buflen = sizeof (buf); // размер буфера
					ret = buffer_id_read (buffer_id, start, count, &histories, buf, &buflen);
					if (ret == 0) {
					} else {
						buflen = sprintf (buf, "ERROR") + 1;
					}						
					IP_MsgCascade (hmsg, buf, buflen, IP_NONE, ST_OK);
					IP_MsgInfoReply (&msginfo, hmsg);
					
				} else
				// очистить буфер данных (buffer_id_destroy [id])
				if (!strncmp ((char*) hmsg->msg->data, "buffer_id_destroy", 17)) {
					int buffer_id;
					sscanf ((char*) hmsg->msg->data + 17, "%d", &buffer_id);
					#ifdef DEBUG_PLCDB
					printf ("buffer_id_destroy (id = %d)\n", buffer_id);
					#endif
					ret = buffer_id_destroy (buffer_id, &histories);
					if (ret == 0) {
						buflen = sprintf (buf, "EOK") + 1;
					} else {
						buflen = sprintf (buf, "ERROR") + 1;
					}						
					IP_MsgCascade (hmsg, buf, buflen, IP_NONE, ST_OK);
					IP_MsgInfoReply (&msginfo, hmsg);
					
				} else
				// сбросить все данные на диск
				if (!strncmp ((char*) hmsg->msg->data, "flush", 5)) {
					ret = 0;
					for (i= 0; i< histories.ncfg; i++) {
						ret |= flush (&histories.cfg [i]);						
					}
					if (ret == 0) {
						buflen = sprintf (buf, "EOK") + 1;
					} else {
						buflen = sprintf (buf, "ERROR") + 1;
					}						
					IP_MsgCascade (hmsg, buf, buflen, IP_NONE, ST_OK);
					IP_MsgInfoReply (&msginfo, hmsg);
					
				} else
				// начать новый файл
				if (!strncmp ((char*) hmsg->msg->data, "start_new_file", 14)) {
					ret = 0;
					for (i= 0; i< histories.ncfg; i++) {
						ret |= start_new_file (&histories.cfg [i]);						
					}
					if (ret == 0) {
						buflen = sprintf (buf, "EOK") + 1;
					} else {
						buflen = sprintf (buf, "ERROR") + 1;
					}						
					IP_MsgCascade (hmsg, buf, buflen, IP_NONE, ST_OK);
					IP_MsgInfoReply (&msginfo, hmsg);
					
				} else
				// delete old files
				if (!strncmp ((char*) hmsg->msg->data, "delete_old_files", 16)) {
					double old_time;
					sscanf ((char*) hmsg->msg->data + 16, "%lf", &old_time);
					ret = 0;
					for (i= 0; i< histories.ncfg; i++) {
						ret |= delete_old_files (histories.cfg [i].filedb, old_time);						
					}
					if (ret == 0) {
						buflen = sprintf (buf, "EOK") + 1;
					} else {
						buflen = sprintf (buf, "ERROR") + 1;
					}						
					IP_MsgCascade (hmsg, buf, buflen, IP_NONE, ST_OK);
					IP_MsgInfoReply (&msginfo, hmsg);
					
				} 				
				else {
					printf ("Unsupported SYNC message\n");
					IP_MsgCascade (hmsg, "Unsupported", 12, IP_NONE, ST_ERROR);
					IP_MsgInfoReply (&msginfo, hmsg);
					exit (-1);					
				}								
				break;
			case IP_RAW:
				/* Message sent using QNX Send command, not through the
				 * Cascade DataHub API.  This is also where a proxy would
				 * be delivered if we had set up a proxy within this task,
				 * for example, a periodic timer.  In that case, the proxy
				 * ID would be (pid_t) sender, and we would not Reply to it. */
				IP_MsgInfoReplyRaw (&msginfo, "Unsupported", 12);
				break;
		}
	}

	return EXIT_SUCCESS;
}
